package com.asiainfo.dacp.datastash.broker.tracelog;

import com.asiainfo.dacp.datastash.broker.core.BrokerContext;
import com.asiainfo.dacp.jdbc.JdbcTemplate;
import com.asiainfo.dacp.util.TableUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.GregorianCalendar;
import java.util.List;

/**
 * 日志的服务
 * @author MeiKefu
 * @date 2017-11-30
 */
@Service
public class StashTraceLoggerService {

    private static Logger LOG = LoggerFactory.getLogger(StashTraceLoggerService.class);

    public void persist(List<StashTraceLogVo> logVoList) {
        if(logVoList!=null && logVoList.size()>0){
            JdbcTemplate jdbcTemplate = new JdbcTemplate(TRACE_LOG_DSNAME);

            String sql = "insert into %s(log_level,broker_id,broker_ip,thread_name,unit_code,batch_no,unit_proc_id,task_proc_id,clz_name,event,message,par_key) values(?,?,?,?,?,?,?,?,?,?,?,?)";

            List<Object[]> batchArgs = new ArrayList<>(logVoList.size());
            int parKey = Integer.parseInt(sdf.format(System.currentTimeMillis()));
            for(StashTraceLogVo logVo : logVoList){
                batchArgs.add(new Object[]{
                        logVo.getLevel(),
                        BrokerContext.getInstance().getBrokerId(),
                        BrokerContext.getInstance().getBrokerIp(),
                        logVo.getThreadName(),
                        logVo.getUnitCode(),logVo.getBatchNo(),logVo.getUnitProcId(),logVo.getTaskProcId(),
                        logVo.getClzName(),logVo.getEvent(),logVo.getMessage(),
                        parKey
                });
            }
            jdbcTemplate.batchUpdate(String.format(sql,TRACE_LOG_TABNAME),batchArgs);
        }
    }

    private String TRACE_LOG_TABNAME = "dacp_datastash_trace_log";
    private String TRACE_LOG_DSNAME = "dataSource";

    public void createCdrTable(){
        JdbcTemplate jt = new JdbcTemplate(TRACE_LOG_DSNAME);
        Pair<String,String> pari = getDdl(jt.getDatabaseType());
        String createDdl = pari.getLeft();
        String alterPartition = pari.getRight();
        if(!TableUtils.existTable(TRACE_LOG_TABNAME)){
            LOG.info("创建日志表{},{}",TRACE_LOG_TABNAME,createDdl);
            TableUtils.createTable(createDdl,jt);
        }else{
            crtPartition(alterPartition,jt);
        }
    }

    private DateFormat sdf = new SimpleDateFormat("yyyyMMdd");
    private Pair<String,String> getDdl(String dbType){
        if("mysql".equalsIgnoreCase(dbType)){
            String cdrTable = " create table %s ("
                    +" log_level varchar(8),"
                    +" broker_id varchar(16),"
                    +" broker_ip varchar(16),"
                    +" thread_name varchar(32),"
                    +" unit_code varchar(16),"
                    +" batch_no varchar(16),"
                    +" unit_proc_id varchar(32),"
                    +" task_proc_id varchar(32),"
                    +" clz_name varchar(32),"
                    +" event varchar(64),"
                    +" message varchar(256),"
                    +" create_dt timestamp,"
                    +" par_key int"
                    +" ) partition by list(par_key) ("
                    +" PARTITION p%s VALUES in (%s),PARTITION p%s VALUES in (%s));";
            Calendar calendar = GregorianCalendar.getInstance();
            String dateTime = sdf.format(calendar.getTime());
            calendar.add(Calendar.DAY_OF_MONTH,1);
            String nexeDt = sdf.format(calendar.getTime());
            String createDdl = String.format(cdrTable,TRACE_LOG_TABNAME,dateTime.substring(2),dateTime,nexeDt.substring(2),nexeDt);
            String alterPartition = String.format("alter table %s add partition ( PARTITION p%s VALUES in (%s) )",TRACE_LOG_TABNAME,nexeDt.substring(2),nexeDt);
            return Pair.of(createDdl,alterPartition);
        }else{
            LOG.error("[{}]不支持的数据源",dbType);
        }
        return Pair.of("","");
    }

    private void crtPartition(String alterPartition,JdbcTemplate jt){
        LOG.info("创建分区,{}",alterPartition);
        try{
            jt.execute(alterPartition);
        }catch(Exception e){
//			e.printStackTrace();
            LOG.warn("{},创建分区失败,{}",alterPartition,e.getMessage());
        }
    }
}
